package com.cs.config;

import com.cs.code.RpcDecoder;
import com.cs.entity.RpcRequest;
import com.cs.handler.UserServerHandler;
import com.cs.serialize.JSONSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;


@Configuration
public class ZookeeperConfig {

    @Value("${zk.url}")
    private String zkUrl;
    @Value("${rpc.url}")
    private String rpcUrl;
    @Value("${rpc.port}")
    private int rpcPort;
    @Value("${zk.path}")
    private String path;

    @Bean
    public  CuratorFramework getCuratorFramework() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zkUrl)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("my-provider")  // 独立的命名空间 /my-provider
                .build();
        client.start();

        //启动RPC服务
        startServer(rpcUrl,rpcPort);

        //服务端启动完成,注册到zookeeper中
        String s = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL).forPath(path, (rpcUrl+":"+rpcPort+",0").getBytes());//初始化节点服务响应时间为0作为最优选择
        System.out.println("注册到zookeeper中成功，节点为:" + s);
        return client;
    }

    @PreDestroy
    private void destroyClient() throws Exception {
        getCuratorFramework().close();
        System.out.println("关闭zookeeper连接成功！");
    }

    //hostName:ip地址  port:端口号
    public void startServer(String hostName,int port) throws Exception {

        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //读写超时皆5sec
                        pipeline.addLast("idleStateHandler",new IdleStateHandler(5, 5, 5, TimeUnit.SECONDS));
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new UserServerHandler());
                    }
                });
        serverBootstrap.bind(hostName,port).sync();
    }
}
